package com.example.waitdemo.utils;

import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;
import com.google.common.collect.Multimaps;
import lombok.extern.log4j.Log4j2;

import java.util.UUID;
import java.util.concurrent.*;

/**
 * TODO   一部调用线程，等待一段时间后返回值
 *         有两部分 1：外网   2  内网， 通过kafak进行通讯
 *         流程：1. 外网 向 kafaka 发送 http 请求 ，kafaka 返回 200 表示知道
 *              2. kafaka 向 内网 发送 http 请求获取查询结果
 *              3. kafaka 收到 内网 返回的结果 向 外网 进行通知
 *              4. 外网 收到结果后返回 到对应的用户
 *     外网 ------------> kafaka -----------> 内网
 *      ^                                      |
 *      |                               五分钟后查询到内容
 *      |                                      |
 *     外网 <-----------  kafaka <------------
 *
 * @ClassName WaitUtil
 * @Author xiaojiang
 * @Date 2021/10/21 下午 17:43
 */
@Log4j2
public class WaitUtil {

//    private static Object obj = new Object();
    // 多线程使用，线程安全的
    private static ConcurrentHashMap<String, Object> effective = new ConcurrentHashMap<>();
    // guava 提供的多值 Map，一个 key 可以对应多个 value
    private static Multimap<String, Object> dataIdContext = Multimaps.synchronizedSetMultimap(HashMultimap.create());


    private static ThreadPoolExecutor threadPool
            // 核心池大小      最大线程数          保持活动时间       时间单位            阵列阻塞数                         线程池执行器            丢弃政策
            = new ThreadPoolExecutor(10, 15, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue(10), new ThreadPoolExecutor.DiscardPolicy());

    public String method(String userId) {

        //异步计算
        FutureTask<String> thread = new FutureTask<String>(new Callable<String>() {
            @Override
            public String call() throws Exception {
                return person(userId);
            }
        });
        threadPool.submit(thread);
        String s = "";
        try {
            // FutureTask.get() 获取返回值
            s = thread.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        return s;
    }

    /**
     *  TODO  向Kafaka进入等待
     * @Param : userId
     * @return String
     * @Author 江
     * @Date 2021/10/21 下午 17:45
     */
    public String person(String userId) {
        String str = UUID.randomUUID().toString().replaceAll("-", "");
        effective.put(str,str);

        // 向kafaka发送http请求（省略）
        // userId 作为kafaka的查询参数
        // str    作为kafaka返回数据时候的对应值
        log.info(userId + "进入线程等待===" + System.currentTimeMillis());
        // 发送成功(返回200)之后进入线程wait

        try {
            System.out.println("当前eff为："+effective);
            Object o = effective.get(str);
            System.out.println("jiasuo"+o);
            synchronized (o) {
                o.wait();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        log.info(userId + "等待结束---" + System.currentTimeMillis());
        String o = String.valueOf(effective.get(str));
        effective.remove(str);
        System.out.println(">>>>>>>>>>>>>>获取参数：" + o + ">>>>>>>>>>>>>>>>>>>>>");

        return o;
    }

    public  void startUdpServer(String userDao) {
        threadPool.submit(new Runnable() {
            @Override
            public void run() {
                // 此处是业务，根据卡发卡传过来的userDao，找到对应的userId,(此处省略，默认userDao就是userId)
                System.out.println("effde 数据是="+effective);
                Object o = effective.get(userDao);
                log.info("--准备将数据"+userDao+"写入 effective--" + System.currentTimeMillis());
                //将数据写入 effective
                effective.put(userDao,"fanhuizhi："+ userDao);
                //唤醒线程
                log.info("--准备唤醒"+userDao+"的线程--" + System.currentTimeMillis());
                synchronized (o) {
                    o.notify();
                }

            }

        });

    }



}
